package com.ruowei.modules.webSocket.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.querydsl.jpa.impl.JPAQueryFactory;
import com.ruowei.config.SpringUtil;
import com.ruowei.modules.sys.repository.SysUserRepository;
import com.ruowei.modules.sys.service.UserJWTService;
import com.ruowei.modules.webSocket.domain.Message;
import com.ruowei.modules.webSocket.domain.MessagePush;
import com.ruowei.modules.webSocket.domain.QMessage;
import com.ruowei.modules.webSocket.domain.QMessagePush;
import com.ruowei.modules.webSocket.domain.enumeration.SenderType;
import com.ruowei.modules.webSocket.pojo.MessageDecoder;
import com.ruowei.modules.webSocket.pojo.MessageEncoder;
import com.ruowei.modules.webSocket.repository.MessagePushRepository;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@ServerEndpoint(value = "/imserver/{userId}", encoders = {MessageEncoder.class}, decoders = {MessageDecoder.class})
@Component
@Slf4j
public class WebSocketService {
    //静态变量，用来记录当前在线连接数
    private static int onlineCount = 0;
    //concurrent包的线程安全set，用来存放每个客户端对应的WebSocket对象
    private static ConcurrentHashMap<String, WebSocketService> webSocketMap = new ConcurrentHashMap<>();
    private Session session;
    private final JPAQueryFactory jpaQueryFactory = SpringUtil.getBean(JPAQueryFactory.class);
    private final MessagePushRepository messagePushRepository = SpringUtil.getBean(MessagePushRepository.class);
    private final SysUserRepository sysUserRepository = SpringUtil.getBean(SysUserRepository.class);
    private final UserJWTService userJWTService = SpringUtil.getBean(UserJWTService.class);

    /**
     * 连接建立成功调用的方法
     *
     * @param session
     * @param userId
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        this.session = session;
        if (webSocketMap.containsKey(userId)) {
            webSocketMap.remove(userId);
            //加入set中
            webSocketMap.put(userId, this);
        } else {
            //加入set中
            webSocketMap.put(userId, this);
            //在线数加1
            addOnlineCount();
        }
        sysUserRepository.findOneByUserCode(userId)
            .ifPresent(userJWTService::online);
        log.info("用户连接:" + userId + ",当前在线人数为:" + getOnlineCount());

        QMessage qMessage = QMessage.message;
        QMessagePush qMessagePush = QMessagePush.messagePush;

        List<Message> unPushedMsg = jpaQueryFactory.selectFrom(qMessage)
            .leftJoin(qMessagePush).on(qMessage.messageCode.eq(qMessagePush.messageCode))
            .where(qMessage.receiverCode.eq(userId)
                .and(qMessagePush.pushed.eq(false)))
            .fetch();
        for (Message message : unPushedMsg) {
            try {
                if (sendMessage(message, userId)) {
                    List<MessagePush> pushes = new ArrayList<>();
                    jpaQueryFactory.selectFrom(qMessagePush)
                        .where(qMessagePush.messageCode.eq(message.getMessageCode()))
                        .fetch()
                        .forEach(messagePush -> {
                            messagePush.setPushed(true);
                            messagePush.setPushTime(Instant.now());
                            pushes.add(messagePush);
                        });
                    messagePushRepository.saveAll(pushes);
                }
            } catch (IOException e) {
                log.error("用户:{}网络异常!", userId);
            }
        }

        try {
            sendMessage("连接成功");
        } catch (IOException e) {
            log.error("用户:{}网络异常!", userId);
        }
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(@PathParam("userId") String userId) {
        if (webSocketMap.containsKey(userId)) {
            webSocketMap.remove(userId);
            //从set中删除
            subOnlineCount();
        }
        sysUserRepository.findOneByUserCode(userId)
            .ifPresent(userJWTService::offline);
        log.info("用户{}退出,当前在线人数为{}", userId, getOnlineCount());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message
     * @param session
     */
    @OnMessage
    public void onMessage(String message, Session session, @PathParam("userId") String userId) throws IOException {
        log.info("用户{}消息,报文:{}", userId, message);
        if (!StringUtils.isEmpty(message)) {
            JSONObject jsonObject = JSON.parseObject(message);
            jsonObject.put("senderType", SenderType.CALLBACK);
            if (StringUtils.isNotEmpty(userId) && webSocketMap.containsKey(userId)) {
                webSocketMap.get(userId).sendMessage(jsonObject.getInnerMap().toString());
            } else {
                log.error("用户{}不在线", userId);
            }
        }
    }

    @OnError
    public void onError(Session session, Throwable error, @PathParam("userId") String userId) {
        log.error("用户错误:{},原因:{}", userId, error.getMessage());
        sysUserRepository.findOneByUserCode(userId)
            .ifPresent(userJWTService::offline);
        error.printStackTrace();
    }

    /**
     * 实现服务器主动推送
     *
     * @param message
     * @throws IOException
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    /**
     * 实现服务器主动推送
     *
     * @param object
     * @throws IOException
     */
    public void sendObject(Object object) throws IOException {
        try {
            this.session.getBasicRemote().sendObject(object);
        } catch (EncodeException e) {
            e.printStackTrace();
        }
    }

    /**
     * 发送自定义消息
     *
     * @param message
     * @param userId
     * @throws IOException
     */
    public static void sendInfo(String message, @PathParam("userId") String userId) throws IOException {
        log.info("发送消息到:{},报文:{}", userId, message);
        if (StringUtils.isNotEmpty(userId) && webSocketMap.containsKey(userId)) {
            webSocketMap.get(userId).sendMessage(message);
        } else {
            log.error("用户{}不在线", userId);
        }
    }

    public static boolean sendMessage(Message message, @PathParam("userId") String userId) throws IOException {
        log.info("发送消息到:{},报文:{}", userId, message);
        if (StringUtils.isNotEmpty(userId) && webSocketMap.containsKey(userId)) {
            webSocketMap.get(userId).sendObject(message);
            return true;
        } else {
            log.error("用户{}不在线", userId);
            return false;
        }
    }

    /**
     * 给一批人发送同一消息
     *
     * @param message
     * @param userIdList
     * @throws IOException
     */
    public static void batchSendMessage(Message message, List<String> userIdList) throws IOException {
        for (String userId : userIdList) {
            log.info("发送消息到:{},报文:{}", userId, message);
            if (StringUtils.isNotEmpty(userId) && webSocketMap.containsKey(userId)) {
                webSocketMap.get(userId).sendObject(message);
            } else {
                log.error("用户{}不在线", userId);
            }
        }

    }

    /**
     * 群发消息
     *
     * @param message
     */
    public static void sendAll(String message) {
        for (Map.Entry<String, WebSocketService> entry : webSocketMap.entrySet()) {
            try {
                entry.getValue().sendMessage(message);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        WebSocketService.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        WebSocketService.onlineCount--;
    }
}
